/**
 * 
 */
package cc.rico.tutorial.workqueue;

import java.io.IOException;
import java.util.concurrent.*;

import cc.rico.tutorial.RabbitMQConnectionUtil;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import lombok.extern.slf4j.Slf4j;

/**
 * @author rico 2017年6月26日
 *
 */
@Slf4j
public class Worker {
	private static final String TASK_QUEUE_NAME = "task_queue";
	
	/**
	 * @param args
	 * @throws IOException 
	 * @throws TimeoutException 
	 */
	public static void main(String[] args) {

		CyclicBarrier barrier = new CyclicBarrier(2, () -> log.info("works started ..."));

		Worker w1 = new Worker("W1");
		Worker w2 = new Worker("W2");

		ExecutorService executor = Executors.newFixedThreadPool(2);
		executor.execute(w1.worker(barrier));
		executor.execute(w2.worker(barrier));
		executor.shutdown();
	}

	private String workerName;

	public Worker(String workerName) {
		this.workerName = workerName;
	}

	public void consume() throws IOException, TimeoutException, InterruptedException {
		Connection conn = RabbitMQConnectionUtil.getConnection();
		Channel channel = conn.createChannel();

		// queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
		channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
		log.info("Waiting for messages. To exit press CTRL+C");

		// Request a specific prefetchCount "quality of service" settings for this channel.
		channel.basicQos(1);
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				log.info("[{}] Received message={}", workerName, message);
				channel.basicAck(envelope.getDeliveryTag(), false);
			}
		};

		// String queue, boolean autoAck, Consumer callback
		channel.basicConsume(TASK_QUEUE_NAME, false, consumer);

		Thread.sleep(1000 * 10);
		RabbitMQConnectionUtil.close(conn, channel);
	}

	public Runnable worker(CyclicBarrier barrier) {
		return () -> {
			try {
				log.info("[{}] waiting for work...", workerName);
				barrier.await();

				log.info("[{}] is working...", workerName);
				this.consume();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			} catch (TimeoutException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}
		};
	}


}
